package cn.caplike.demo.kafka.producer.ack;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Date;
import java.util.Properties;

/**
 * acks=0，生产者发送过来数据就不管了，可靠性差，效率高
 * <br>
 * acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等
 * <br>
 * acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follower应答，可靠性高，效率低
 *
 * @author <a href="mailto:like@pystandard.com">LiKe</a> @ 2023-09-02 16:12:15
 */
@SuppressWarnings("SpellCheckingInspection")
public class ProducerAck {

    public static void main(String[] args) {
        // 配置
        final Properties properties = new Properties();

        // 连接集群 bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "big-data-1:9092,big-data-2:9092,big-data-3:9092");
        // 指定对应 key 和 value 的序列化类型 key.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // acks
        properties.put(ProducerConfig.ACKS_CONFIG, "1");
        // 重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);

        // 创建生产者对象
        try (final KafkaProducer<String, String/* "", hello */> producer = new KafkaProducer<>(properties)) {
            // 普通异步发送数据
            producer.send(new ProducerRecord<>("first", 0, String.valueOf(new Date().getTime()), "caplike@"+ new Date().getTime())); // 会自动创建 topic, 发送一条消息, 所有消费者都能收到
        }
    }
}
